package com.ry.flink.jobx;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

import java.time.LocalDateTime;

/**
 *  9. 表的Join操作
 * */
public class Job9 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tablEnv = StreamTableEnvironment.create(env);

        /**
         * userStream
         * */
        DataStream<Row> userStream = env.fromElements(
                Row.of(LocalDateTime.parse("2022-05-06T13:00:00"), 1, "大乔"),
                Row.of(LocalDateTime.parse("2022-05-06T13:00:00"), 2, "貂蝉"),
                Row.of(LocalDateTime.parse("2022-05-06T13:00:00"), 2, "貂蝉"))
                .returns(
                        Types.ROW_NAMED(
                                new String[]{"ts", "uid", "name"},
                                Types.LOCAL_DATE_TIME, Types.INT, Types.STRING
                        ));

        DataStream<Row> orderStream = env.fromElements(
                Row.of(LocalDateTime.parse("2022-05-06T13:02:00"), 1, 500),
                Row.of(LocalDateTime.parse("2022-05-06T13:06:00"), 2,700),
                Row.of(LocalDateTime.parse("2022-05-06T13:012:00"), 2, 3000))
                .returns(
                        Types.ROW_NAMED(
                                new String[]{"ts", "uid", "amount"},
                                Types.LOCAL_DATE_TIME, Types.INT, Types.STRING
                        ));

        //userTable
        tablEnv.createTemporaryView(
                "userTable",
                userStream,
                Schema.newBuilder()
                        .column("ts", DataTypes.TIMESTAMP(3))
                        .column("uid", DataTypes.INT())
                        .column("name", DataTypes.STRING())
                .watermark("ts","ts - INTERVAL '2' SECOND")
                .build());

        //orderTable
        tablEnv.createTemporaryView(
                "orderTable",
                userStream,
                Schema.newBuilder()
                        .column("ts", DataTypes.TIMESTAMP(3))
                        .column("uid", DataTypes.INT())
                        .column("amount", DataTypes.INT())
                        .watermark("ts","ts - INTERVAL '2' SECOND")
                        .build());



        /**
         * orderStream
         * */
        env.execute("job9");

    }
}
